package com.hexinfo.dmpro.sparing.service.impl;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.date.DateField;
import cn.hutool.core.date.DateTime;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.io.file.FileAppender;
import cn.hutool.core.util.StrUtil;
import com.hexinfo.dmpro.common.model.MessageCenter;
import com.hexinfo.dmpro.common.service.MessageCenterService;
import com.hexinfo.dmpro.common.utils.ScanCommonConstants;
import com.hexinfo.dmpro.sparing.dao.SyncRuleConfMapper;
import com.hexinfo.dmpro.sparing.dao.TimeFieldMatchingMapper;
import com.hexinfo.dmpro.sparing.dto.*;
import com.hexinfo.dmpro.sparing.model.TimeFieldMatching;
import com.hexinfo.dmpro.sparing.service.*;
import com.hexinfo.dmpro.sparing.util.Logger;
import com.xqfunds.job.core.log.XxlJobLogger;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import javax.sql.DataSource;
import java.io.File;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.Statement;
import java.util.*;
import java.util.concurrent.Future;

/**
 * 删除日期分区数据，及目标集群有、源集群没有的日期分区数据
 */
@Service
@Slf4j
public class DateDataServiceImpl implements DateDataService {

    @Value("${conn.pool.impala.source.url}")
    private String souUrl;
    @Value("${conn.pool.impala.target.url}")
    private String tarUrl;
    @Value("${hdfs.tarPrefix}")
    private String tarPrefix;     //同步删除，目标库前缀

    @Autowired
    private SyncRuleConfMapper syncRuleConfMapper;
    @Autowired
    private TimeFieldMatchingMapper timeFieldMatchingMapper;
    @Autowired
    private AsyncDateDataService asyncDateDataService;
    @Autowired
    private LsFilterSouService lsFilterSouService;
    @Autowired
    private FutureService futureService;

    /**
     * 删除日期分区数据，及目标集群有、源集群没有的日期分区数据
     */
    @Override
    public StateDataDTO dropDateData(String source,FileAppender heatlog, List<WhereScriptDTO> whereScriptDTOList) {
        heatlog.append(logTime("-----------------【同步删除日期分区数据开始】------------------------"));
        heatlog.append(logTime("【源的】impala的url："+souUrl));
        heatlog.append(logTime("【目标】impala的url："+tarUrl));

        //按源值获取包含查询所有日期分区sql的 List<DateDataDTO>
//        List<DateDataDTO> dateDataDTOs = getDateDataDTOs(source,heatlog);
        List<DateDataDTO> dateDataDTOs = getDateDataDTOsWhere(source,heatlog,whereScriptDTOList);
        int sizes = dateDataDTOs.size();
        heatlog.append(logTime("【含日期分区的表数量："+sizes+"】"));

        //给一个任务执行情况初始状态
        asyncDateDataService.updateTaskState(source,true);
        StateDataDTO stateDataDTO = new StateDataDTO();
        if (sizes > 0){
//            //结束状态，给一个初始值
//            asyncDateDataService.updateEndFlag(source,false);

            // 获取过滤后的每个表的hdfs相关信息List集合
            List<TableDistCPDTO> tableDistCPDTOs = lsFilterSouService.getTableDistCPDTOMap(source);
            heatlog.append(logTime("【同步删除--LS过滤后的表数量】："+tableDistCPDTOs.size()));
            heatlog.flush();
            String logPath = tableDistCPDTOs.get(0).getLogPath();
            String tbLogPath = logPath.substring(0, logPath.lastIndexOf("/"));
            //线程池异步,通过源和目标select查询出目标中比源多的日期分区，目标执行dropSql，刷新目标表
            List<Future<Boolean>> futures = new ArrayList<>();
            for (DateDataDTO dateDataDTO : dateDataDTOs) {
                //实体类添加相应的，表级日志分析记录表，的id
                for (TableDistCPDTO tableDistCPDTO : tableDistCPDTOs) {
                    if (dateDataDTO.getDbTbName().equals(tableDistCPDTO.getDbTableName())){
                        dateDataDTO.setTableId(tableDistCPDTO.getTableId());
                        break;
                    }
                }
                Future<Boolean> booleanFuture = asyncDateDataService.execSql(dateDataDTO, source, sizes, heatlog, tbLogPath);
                futures.add(booleanFuture);
            }
            //判断全部结束
            stateDataDTO = futureService.futureStatusList(futures, heatlog,"同步删除");
            //----------------------------循环判断删除任务是否全部执行完---------------
//            //循环判断这个源的热备是否全部结束
//            while (!asyncDateDataService.getEndFlag(source)){
//                try {
//                    Thread.sleep(5000);
//                } catch (InterruptedException e) {
//                    e.printStackTrace();
//                }
//            }
//            //删除源key
//            asyncDateDataService.deleteEndFlag(source);
        }else {
            stateDataDTO.setDataList(new ArrayList<>());
        }
        heatlog.append(logTime("-----------------【同步删除日期分区数据执行完成】------------------------"));
        heatlog.flush();

        //----------------------------判断删除任务执行是否异常---------------
        if(asyncDateDataService.getTaskState(source)){
            stateDataDTO.setState(true);
            asyncDateDataService.deleteTaskState(source);  //删除key
        }else {
            stateDataDTO.setState(false);
            asyncDateDataService.deleteTaskState(source);   //删除key
        }
        return stateDataDTO;
    }

    /**
     * 获取包含查询所有日期分区sql的 List<DateDataDTO>
     */
    private List<DateDataDTO> getDateDataDTOsWhere(String source,FileAppender heatlog,List<WhereScriptDTO> whereScriptDTOList) {
        List<DateDataDTO> dateDataDTOs = new ArrayList<>();
        String dropSql = "alter table {} drop if exists partition ({} cast({} as int) < {})";
        String souSql = "SELECT {} FROM {} where 1=1 {} and cast({} as int) between {} and {} GROUP BY {}";
        String tarSql = "SELECT {} FROM {} where 1=1 {} GROUP BY {}";
        String souCol = "and cast({} as string) = '{}'";
        String delCol = "cast({} as string) = '{}' ,";
        DateTime date = DateUtil.date();  //获取当前时间
        heatlog.append(logTime("当前时间："+date));
        //获取每个表的sql
        for (WhereScriptDTO dto : whereScriptDTOList) {
            String nowDate = "";    //对应格式当前时间
            String twoDate = "";  //对应格式同步范围前时间
            if(StrUtil.isNotBlank(dto.getSourceQueryCriteria())){
                if(dto.getSourceQueryCriteria().contains("as int")){
                    DateDataDTO dateDataDTO = new DateDataDTO();
                    dateDataDTO.setDbName(dto.getSourceDbName());
                    dateDataDTO.setTbName(dto.getSourceTableName());

                    String str = dto.getSourceQueryCriteria().replaceAll("where","");
                    String str1 = str.replaceAll("cast\\(","").replaceAll("\\)","");

                    if(str1.contains("as string")){
                        String[] str2 = str1.split("as string");
                        dateDataDTO.setSouColName(str2[0].replaceAll(" ",""));
                        //判断是否包含as int
                        String[] str3 = str2[1].split("as int");
                        String[] str4 = str3[0].split("and");
                        String str5 = str4[0].replaceAll("\\'","").replaceAll("=","").replaceAll(" ","");
                        dateDataDTO.setSource(str5);
                        dateDataDTO.setDateColName(str4[1].replaceAll(" ",""));
                        String[] str6 = str3[1].replaceAll("BETWEEN ","").split("and");
                        twoDate = str6[0].replaceAll(" ","");
                        nowDate = str6[1].replaceAll(" ","");
                    }else{
                        String[] str3 =str1.split("as int");
                        dateDataDTO.setDateColName(str3[0].replaceAll(" ",""));
                        String[] str6 = str3[1].replaceAll("BETWEEN ","").split("and");
                        twoDate = str6[0].replaceAll(" ","");
                        nowDate = str6[1].replaceAll(" ","");
                    }
                    String sou = "";
                    String tar = "";
                    String tarDropSql = "";
                    //判断有没有源分区
                    if (StrUtil.isNotBlank(dateDataDTO.getSouColName())) {
                        String delColSql = StrUtil.format(delCol, dateDataDTO.getSouColName(), dateDataDTO.getSource());
                        String souColSql = StrUtil.format(souCol, dateDataDTO.getSouColName(), dateDataDTO.getSource());
                        tarDropSql = StrUtil.format(dropSql, tarPrefix+dateDataDTO.getDbTbName(), delColSql,dateDataDTO.getDateColName(),twoDate);
                        sou = StrUtil.format(souSql, dateDataDTO.getDateColName(), dateDataDTO.getDbTbName(), souColSql,
                                dateDataDTO.getDateColName(), twoDate, nowDate, dateDataDTO.getDateColName());
                        tar = StrUtil.format(souSql, dateDataDTO.getDateColName(), tarPrefix+dateDataDTO.getDbTbName(), souColSql,
                                dateDataDTO.getDateColName(), twoDate, nowDate, dateDataDTO.getDateColName());
//                        tar = StrUtil.format(tarSql, dateDataDTO.getDateColName(), tarPrefix+dateDataDTO.getDbTbName(), souColSql,
//                                dateDataDTO.getDateColName());
                    }else {
                        tarDropSql = StrUtil.format(dropSql, tarPrefix+dateDataDTO.getDbTbName(), "",dateDataDTO.getDateColName(),twoDate);
                        sou = StrUtil.format(souSql, dateDataDTO.getDateColName(), dateDataDTO.getDbTbName(), "",
                                dateDataDTO.getDateColName(), twoDate, nowDate, dateDataDTO.getDateColName());
                        tar = StrUtil.format(souSql, dateDataDTO.getDateColName(), tarPrefix+dateDataDTO.getDbTbName(), "",
                                dateDataDTO.getDateColName(), twoDate, nowDate, dateDataDTO.getDateColName());
//                        tar = StrUtil.format(tarSql, dateDataDTO.getDateColName(), tarPrefix+dateDataDTO.getDbTbName(), "",
//                                dateDataDTO.getDateColName());
                    }
                    dateDataDTO.setSouSql(sou);
                    dateDataDTO.setTarSql(tar);
                    dateDataDTO.setDelSql(tarDropSql);
                    dateDataDTOs.add(dateDataDTO);
                }
            }
        }
        return dateDataDTOs;
    }


    /**
     * 获取包含查询所有日期分区sql的 List<DateDataDTO>
     */
    private List<DateDataDTO> getDateDataDTOs(String source,FileAppender heatlog) {
        //查询规则-元数据视图表获取数据
        List<DateDataDTO> dateDataDTOs = syncRuleConfMapper.getDateDataDTOs(source);

        //日期分区格式表中所有日期分区字段名
        List<TimeFieldMatching>  dateNameTypess = timeFieldMatchingMapper.getDateNameType();
        Map<String,String> dateNameTypes = new HashMap<>();
        for (TimeFieldMatching dateNameType : dateNameTypess) {
            dateNameTypes.put(dateNameType.getTimeFieldKay(),dateNameType.getTimeFieldValue());
        }
        DateTime date = DateUtil.date();  //获取当前时间
        heatlog.append(logTime("当前时间："+date));
//        Logger.log("当前时间："+date);
//        String souSql = "SELECT {dateColName} FROM {dbTbName} where 1=1 {souCol} and cast({dateColName} as int) between {twoDate} and {nowDate} GROUP BY {dateColName}";
//        String tarSql = "SELECT {dateColName} FROM {dbTbName} where 1=1 {souCol} GROUP BY {dateColName}";
//        String souCol = "and cast({souColName} as string) = '{souName}'";
        String souSql = "SELECT {} FROM {} where 1=1 {} and cast({} as int) between {} and {} GROUP BY {}";
        String tarSql = "SELECT {} FROM {} where 1=1 {} GROUP BY {}";
        String souCol = "and cast({} as string) = '{}'";
        //获取每个表的sql
        for (DateDataDTO dateDataDTO : dateDataDTOs) {
            //获取日期分期格式（yyyymmdd、yyyymm、yyyy）
            String dateType = dateNameTypes.get(dateDataDTO.getDateColName());
            int betweenDay = -dateDataDTO.getSyncRange();
            //月判断
            if(dateType.length()==6){
                betweenDay = -24;
            //年判断
            }else if(dateType.length()==4){
                betweenDay = -2;
            }
            // 获取同步范围前的日期
            DateTime two = DateUtil.offset(date, DateField.DAY_OF_MONTH, betweenDay);
            String nowDate = DateUtil.format(date,dateType);    //对应格式当前时间
            String twoDate = DateUtil.format(two,dateType);  //对应格式同步范围前时间
            String sou = "";
            String tar = "";
            //判断有没有源分区
            if (StrUtil.isNotBlank(dateDataDTO.getSouColName())) {
                String souColSql = StrUtil.format(souCol, dateDataDTO.getSouColName(), source);
                sou = StrUtil.format(souSql, dateDataDTO.getDateColName(), dateDataDTO.getDbTbName(), souColSql,
                        dateDataDTO.getDateColName(), twoDate, nowDate, dateDataDTO.getDateColName());
                tar = StrUtil.format(tarSql, dateDataDTO.getDateColName(), tarPrefix+dateDataDTO.getDbTbName(), souColSql,
                        dateDataDTO.getDateColName());
            }else {
                sou = StrUtil.format(souSql, dateDataDTO.getDateColName(), dateDataDTO.getDbTbName(), "",
                        dateDataDTO.getDateColName(), twoDate, nowDate, dateDataDTO.getDateColName());
                tar = StrUtil.format(tarSql, dateDataDTO.getDateColName(), tarPrefix+dateDataDTO.getDbTbName(), "",
                        dateDataDTO.getDateColName());
            }
            dateDataDTO.setSouSql(sou);
            dateDataDTO.setTarSql(tar);
        }
        return dateDataDTOs;
    }

    /**
     * 日志加日期时间
     */
    private String logTime(String str) {
        return "["+DateUtil.format(DateUtil.date(), "yyyy-MM-dd HH:mm:ss")+"]" +str;
    }

}
